package com.mtw.bbs.reply.job;

import com.alibaba.fastjson.JSONObject;
import com.mtw.bbs.common.core.constant.MQConstant;
import com.mtw.bbs.common.core.util.ExecutorUtilz;
import com.mtw.bbs.constant.ReplyConstant;
import com.mtw.bbs.reply.mapper.ReplyIndexMapper;
import com.mtw.bbs.reply.mapper.ReplyInfoMapper;
import com.mtw.bbs.reply.mapper.ReplySubjectMapper;
import com.mtw.bbs.reply.pojo.entity.ReplyIndex;
import com.mtw.bbs.reply.pojo.entity.ReplyInfo;
import com.mtw.bbs.reply.pojo.form.ReplyForm;
import com.mtw.bbs.common.redisBase.util.RedisUtilz;
import jakarta.annotation.Resource;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.context.annotation.Lazy;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CompletableFuture;

/**
 * 异步更新任务
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ReplyAsyncJob {


    private final ExecutorUtilz executorUtilz;
    private final RedissonClient redissonClient;
    private final ReplyInfoMapper replyInfoMapper;
    private final RocketMQTemplate rocketMQTemplate;
    private final ReplyIndexMapper replyIndexMapper;
    private final ThreadPoolTaskExecutor asyncExecutor;
    private final RedisUtilz<String,Object> redisUtilz;
    private final ReplySubjectMapper replySubjectMapper;


    @Lazy
    @Resource
    private ReplyAsyncJob owner;





    /**
     * 更新评论数量
     * @param form              当前评论
     * @param increase          评论增量
     */
    public void updateReplyCount(ReplyForm form, Integer increase){
        CompletableFuture.runAsync(()->{

            String lockKey = ReplyConstant.LOCK_REPLY_RECORD_PREFIX + form.getId();
            RLock lock = redissonClient.getLock(lockKey);
            lock.lock();
            try {
                String redisKey = ReplyConstant.REDIS_REPLY_COUNT_PREFIX + form.getObjectType() + "_" + form.getObjectId();
                redisUtilz.delete(redisKey);
                ReplyIndex index = replyIndexMapper.selectById(form.getId());
                if (index != null){
                    form.setRootId(index.getRootId());
                    form.setParentId(index.getParentId());
                    owner.doUpdateReplyCount(form,increase);
                    // 延时删除
                    executorUtilz.delayedTask(()->redisUtilz.delete(redisKey));
                }
            }finally {
                lock.unlock();
            }
        },asyncExecutor);
    }



    /**
     * 停用评论时更新评论数量
     * @param form              当前评论
     * @param isEnable          是否为启用，true为启用
     */
    public void updateReplyCountByStatus(ReplyForm form, boolean isEnable){
        CompletableFuture.runAsync(()->{

            String lockKey = ReplyConstant.LOCK_REPLY_RECORD_PREFIX + form.getId();
            RLock lock = redissonClient.getLock(lockKey);
            lock.lock();
            try {
                String redisKey = ReplyConstant.REDIS_REPLY_COUNT_PREFIX + form.getObjectType() + "_" + form.getObjectId();
                redisUtilz.delete(redisKey);
                ReplyInfo replyInfo = replyInfoMapper.selectById(form.getId());
                if (replyInfo != null){
                    int count = Math.toIntExact(replyInfo.getReplyCount());
                    owner.doUpdateReplyCount(form,isEnable?count:-count);
                    // 延时删除
                    executorUtilz.delayedTask(()->redisUtilz.delete(redisKey));
                }
            }finally {
                lock.unlock();
            }
        },asyncExecutor);
    }




    /**
     * 删除评论时更新评论数量
     * @param form              当前评论
     */
    public void updateReplyCountByDelete(ReplyForm form){

        CompletableFuture.runAsync(()->{
            String lockKey = ReplyConstant.LOCK_REPLY_RECORD_PREFIX + form.getId();
            RLock lock = redissonClient.getLock(lockKey);
            lock.lock();
            try {
                String redisKey = ReplyConstant.REDIS_REPLY_COUNT_PREFIX + form.getObjectType() + "_" + form.getObjectId();
                redisUtilz.delete(redisKey);
                ReplyInfo replyInfo = replyInfoMapper.selectById(form.getId());
                if (replyInfo != null){
                    int count = Math.toIntExact(replyInfo.getReplyCount());
                    owner.doUpdateReplyCount(form,-count);
                    // 延时删除
                    executorUtilz.delayedTask(()->redisUtilz.delete(redisKey));
                }
            }finally {
                lock.unlock();
            }
        },asyncExecutor);
    }



    /**
     * 更新评论数量
     * @param form      表单
     * @param increase  评论增量
     */
    @Transactional
    public void doUpdateReplyCount(ReplyForm form,Integer increase){

        // 存在父评论
        if (!ReplyConstant.DEFAULT_ROOT_ID.equals(form.getParentId())){
            Set<String> ids = new HashSet<>(2);
            ids.add(form.getParentId());
            ids.add(form.getRootId());
            replyInfoMapper.updateReplyCount(increase, ids);
        }
        // 更新总评论数量
        replySubjectMapper.updateReplyCount(increase,form.getObjectId(),form.getObjectType());

        // 发送mq消息
        sendMQ(form);

    }


    public void sendMQ(ReplyForm form){
        try{
            log.error("发送MQ,{}",form);
            rocketMQTemplate.syncSend(MQConstant.REPLY_TOPIC+":"+form.getObjectType(), MessageBuilder.withPayload(JSONObject.toJSON(form)).build());
            // MQ似乎有点多余
        }catch (Exception e){
            log.error("发送MQ异常：",e);
        }

    }
}
